package com.yyf.kafka.producer;

import com.yyf.kafka.dto.MessageEntity;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * 壹 * @Author: yang
 * 貮 * @Date: 2019/11/5 14:40
 */
@Component
public class SimpleProducer {

	@Autowired
	private KafkaTemplate<String,MessageEntity> kafkaTemplate;

	public void send(String topic,MessageEntity message){
		kafkaTemplate.send(topic,message);
	}

	public void send(String topic,String key,MessageEntity entity){

		ProducerRecord<String,MessageEntity> record = new ProducerRecord<>(
				topic,
				key,
				entity
		);

		long startTime = System.currentTimeMillis();

		ListenableFuture<SendResult<String,MessageEntity>> future
				= kafkaTemplate.send(record);
		future.addCallback(new ProducerCallback(startTime, key, entity));

	}
}
